package org.leiyang.mq;

import org.leiyang.common.redis.ReactiveRedisCache;
import org.leiyang.repository.MsgOneToOneRepository;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.stream.StreamListener;
import org.springframework.stereotype.Component;
import reactor.core.publisher.Mono;

import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

import static org.leiyang.common.util.Constants.*;

/**
 * 客户端成功收到消息的消费
 */
@Component("msgReceiveConsumer")
public class MsgReceiveConsumer implements StreamListener<String, MapRecord<String, String, String>> {
    private final Logger logger = LoggerFactory.getLogger(getClass());

    @Value("${successReceiveMsg.consumer.id}")
    private String successReceiveMsgConsumerId;

    private final ReactiveRedisCache redisCache;
    private final MsgOneToOneRepository msgOneToOneRepository;

    public MsgReceiveConsumer(ReactiveRedisCache redisCache, MsgOneToOneRepository msgOneToOneRepository) {
        this.redisCache = redisCache;
        this.msgOneToOneRepository = msgOneToOneRepository;
    }

    @Override
    public void onMessage(MapRecord<String, String, String> message) {
        try {
            Map<String, String> msgMap = message.getValue();
            String msgIdsStr = msgMap.get("msgIds");
            //过滤掉已经处理过的消息
            List<Long> msgIds = Arrays.stream(msgIdsStr.split(",")).map(Long::parseLong).collect(Collectors.toList());
            Map<Long, Boolean> filterRe = redisCache.batchGetBitValues(FILTER_RECE_MSG_IDS, msgIds);
            List<Long> processedMsgIds = filterRe.entrySet().stream().filter(Map.Entry::getValue).map(Map.Entry::getKey).collect(Collectors.toList());
            if(!processedMsgIds.isEmpty()) {
                logger.warn("本批次消息里存在部分消息已经处理过了!");
                msgIds.removeAll(processedMsgIds);
            }
            //批量更新消息的接收状态为已接收
            msgOneToOneRepository.batchUpdateMsgOneToOneReceive(msgIds.stream().map(String::valueOf).collect(Collectors.toList()))
                .flatMap(lines -> {
                    if (lines > 0) {
                        //标记处理过的消息
                        return redisCache.batchSetBitValues(FILTER_RECE_MSG_IDS, msgIds, true).map(re -> lines);
                    }
                    return Mono.just(lines);
                })
                .subscribe(lines -> logger.info("已成功更新{}条消息的接收状态: 已接收.", lines));
        } catch (Exception e) {
            logger.error("消费到MsgReceiveResp:{} 进行业务处理时出现异常, ", message.getValue(), e);
        } finally {
            //始终会消息应答
            redisCache.createStreamOperations().acknowledge(MSG_REC_SUCCESS_TOPIC, MSG_REC_SUCCESS_GROUP_ID, message.getId());
        }
    }
}
